package com.mr.wordcount.compress;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
 * 设置压缩
 * 输入目录中包含deflate,gzip,bzip2,snappy文件时不需要额外设置。包含lzo文件的时候需要额外设置
 *
 * 没使用map输出压缩之前：	Map output materialized bytes=9774693
 * 使用deflate压缩格式：	Map output materialized bytes=328177
 * 使用Bzip2压缩格式：		Map output materialized bytes=369514
 *
 * 没使用reduce输出压缩之前：	621KB
 * 使用deflate压缩格式：		196KB
 * 使用Bzip2压缩格式：			207KB
 */

public class WCDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		// 作为整个job的配置
		Configuration conf = new Configuration();

		// 设置map输出压缩(在shuffle阶段使用压缩，默认使用的压缩格式为deflate)
		conf.set("mapreduce.map.output.compress", "true");
		// 设置使用Bzip2压缩格式
		conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.BZip2Codec");

		// 设置reduce输出压缩
		conf.set("mapreduce.output.fileoutputformat.compress", "true");
		conf.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.BZip2Codec");

		// 创建job
		Job job = Job.getInstance(conf);

		// 设置名称
		job.setJobName("wordcount_compress");

		// 将类WCDriver所在的jar包作为job的jar包
		job.setJarByClass(WCDriver.class);


		// 设置job运行的Mapper,Reducer类
		job.setMapperClass(WCMapper.class);
		job.setReducerClass(WCReducer.class);

		// job需要根据Mapper和Reducer输出的key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化
		// 如果Mapper和Reducer输出的key-value一致，直接设置job最终的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);


		// 设置输入输出目录
		Path input = new Path("E:/尚硅谷大数据/05-Hadoop，HadoopHA，Zookeeper/05-Mapreduce/mrinput/wordcount_compress");
		Path output = new Path("E:/尚硅谷大数据/05-Hadoop，HadoopHA，Zookeeper/05-Mapreduce/mroutput/wordcount_compress");

		// 保证输出目录不能存在
		FileSystem fs = FileSystem.get( conf);

		if (fs.exists(output)) {
			fs.delete(output, true);
		}


		FileInputFormat.setInputPaths(job, input);
		FileOutputFormat.setOutputPath(job, output);

		// 运行job
		job.waitForCompletion(true);
	}

}


